package com.learning.hadoop.sqoop.client;

import org.apache.sqoop.client.SqoopClient;
import org.apache.sqoop.model.MConfig;
import org.apache.sqoop.model.MConnector;
import org.apache.sqoop.model.MDriverConfig;
import org.apache.sqoop.model.MFromConfig;
import org.apache.sqoop.model.MInput;
import org.apache.sqoop.model.MJob;
import org.apache.sqoop.model.MLink;
import org.apache.sqoop.model.MLinkConfig;
import org.apache.sqoop.model.MSubmission;
import org.apache.sqoop.model.MToConfig;
import org.apache.sqoop.submission.counter.Counter;
import org.apache.sqoop.submission.counter.CounterGroup;
import org.apache.sqoop.submission.counter.Counters;
import org.apache.sqoop.validation.Status;

import java.util.Collection;
import java.util.List;
import java.util.ResourceBundle;

/**
 * ClassName: SqoopClient
 * Description: http://sqoop.apache.org/docs/1.99.5/ClientAPI.html
 * Date: 2018/4/18 9:36 【需求编号】
 *
 * @author Sam Sho
 * @version V1.0.0
 */
public class SqoopCli {


    /**
     * Connector
     */
    public void testConnector() {
        String url = "http://hadoop1:12000/sqoop/";
        SqoopClient client = new SqoopClient(url);
        // 这边可以修改
//        client.setServerUrl(url);

        Collection<MConnector> connectors = client.getConnectors();
        connectors.forEach(System.out::println);

        MConnector connector = client.getConnector(1);
        System.out.println(connector.getClassName());
        System.out.println(connector.getUniqueName());
        System.out.println(connector.getPersistenceId());
    }

    /**
     * Link
     */
    public void testLink() {
        String url = "http://hadoop1:12000/sqoop/";
        SqoopClient client = new SqoopClient(url);
        List<MLink> links = client.getLinks();
        links.forEach(System.out::println);

        MLink link = client.getLink(1);
        System.out.println(link.getConnectorId());
        System.out.println(link.getPersistenceId());
        System.out.println(link.getName());

        System.out.println(link.getEnabled());
        MLinkConfig linkConfig = link.getConnectorLinkConfig();
        System.out.println(linkConfig);
        linkConfig.getConfigs().forEach(System.out::println);

    }

    public void testLink2() {
        String url = "http://hadoop1:12000/sqoop/";
        SqoopClient client = new SqoopClient(url);

        // create a placeholder for link
        long connectorId = 1;
        MLink link = client.createLink(connectorId);
        link.setName("db_link");
        link.setCreationUser("hadoop");
        MLinkConfig linkConfig = link.getConnectorLinkConfig();
        // fill in the link config values
        linkConfig.getStringInput("linkConfig.connectionString").setValue("jdbc:mysql://10.1.219.78:3306/qmpnew");
        linkConfig.getStringInput("linkConfig.jdbcDriver").setValue("com.mysql.jdbc.Driver");
        linkConfig.getStringInput("linkConfig.username").setValue("root");
        linkConfig.getStringInput("linkConfig.password").setValue("root");
        // save the link object that was filled
        Status status = client.saveLink(link);
        if (status.canProceed()) {
            System.out.println("Created Link with Link Id : " + link.getPersistenceId());
        } else {
            System.out.println("Something went wrong creating the link");
        }

        client.updateLink(link);
        client.deleteLink(link.getPersistenceId());
    }

    /**
     * hdfs2mysql
     */
    public void createJob() {
        String url = "http://hadoop1:12000/sqoop/";
        SqoopClient client = new SqoopClient(url);
        //Creating dummy job object
        // for jdbc connector
        long fromLinkId = 2;
        // for HDFS connector
        long toLinkId = 1;
        MJob job = client.createJob(fromLinkId, toLinkId);
        job.setName("api_test_job");
        job.setCreationUser("hadoop");
        // set the "FROM" link job config values
        MFromConfig fromJobConfig = job.getFromJobConfig();
        fromJobConfig.getStringInput("fromJobConfig.inputDirectory").setValue("/data/bigdata");
//
        // set the "TO" link job config values
        MToConfig toJobConfig = job.getToJobConfig();
        toJobConfig.getStringInput("toJobConfig.schemaName").setValue("qmpnew");
        toJobConfig.getStringInput("toJobConfig.tableName").setValue("t_test2");
//        fromJobConfig.getStringInput("toJobConfig.partitionColumn").setValue("id");

        // set the driver config values
        MDriverConfig driverConfig = job.getDriverConfig();
        driverConfig.getIntegerInput("throttlingConfig.numExtractors").setValue(50);
        driverConfig.getIntegerInput("throttlingConfig.numLoaders").setValue(50);

        Status status = client.saveJob(job);

        if (status.canProceed()) {
            System.out.println("Created Job with Job Id: " + job.getPersistenceId());
        } else {
            System.out.println("Something went wrong creating the job");
        }
    }

    public void updateJob() {
        String url = "http://hadoop1:12000/sqoop/";
        SqoopClient client = new SqoopClient(url);

        MJob job = client.getJob(4);
        MFromConfig fromJobConfig = job.getFromJobConfig();
        fromJobConfig.getStringInput("fromJobConfig.inputDirectory").setValue("/data/bigdata");

        Status status = client.updateJob(job);
//        client.deleteJob(job.getPersistenceId());
    }

    public void startJob() {
        String url = "http://hadoop1:12000/sqoop/";
        SqoopClient client = new SqoopClient(url);
        long jobId = 4;
        MSubmission submission = client.startJob(jobId);
        System.out.println("Job Submission Status : " + submission.getStatus());
        if (submission.getStatus().isRunning() && submission.getProgress() != -1) {
            System.out.println("Progress : " + String.format("%.2f %%", submission.getProgress() * 100));
        }
        System.out.println("Hadoop job id :" + submission.getExternalJobId());
        System.out.println("Job link : " + submission.getExternalLink());
        Counters counters = submission.getCounters();
        if (counters != null) {
            System.out.println("Counters:");
            for (CounterGroup group : counters) {
                System.out.print("\t");
                System.out.println(group.getName());
                for (Counter counter : group) {
                    System.out.print("\t\t");
                    System.out.print(counter.getName());
                    System.out.print(": ");
                    System.out.println(counter.getValue());
                }
            }
        }
        if (submission.getError() != null) {
            System.out.println("Exception info : " + submission.getError().getErrorDetails());
        }
        //Check job status for a running job
        submission = client.getJobStatus(jobId);
        if (submission.getStatus().isRunning() && submission.getProgress() != -1) {
            System.out.println("Progress : " + String.format("%.2f %%", submission.getProgress() * 100));
        }

        //Stop a running job
//        client.stopJob(jobId);
    }

    public void descConfigAndInputConnector() {
        String url = "http://hadoop1:12000/sqoop/";
        SqoopClient client = new SqoopClient(url);
        long connectorId = 1;
        // link config for connector
        describe(client.getConnector(connectorId).getLinkConfig().getConfigs(), client.getConnectorConfigBundle(connectorId));
        // from job config for connector
        describe(client.getConnector(connectorId).getFromConfig().getConfigs(), client.getConnectorConfigBundle(connectorId));
        // to job config for the connector
        describe(client.getConnector(connectorId).getToConfig().getConfigs(), client.getConnectorConfigBundle(connectorId));
    }

    void describe(List<MConfig> configs, ResourceBundle resource) {
        for (MConfig config : configs) {
            System.out.println(resource.getString(config.getLabelKey()) + ":");
            List<MInput<?>> inputs = config.getInputs();
            for (MInput input : inputs) {
                System.out.println(resource.getString(input.getLabelKey()) + " : " + input.getValue());
            }
            System.out.println();
        }

    }
}
